package org.tksj.bim.event;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Handler;
import io.vertx.core.eventbus.EventBus;
import io.vertx.core.eventbus.Message;
import org.tksj.bim.common.mq.ConsumerHandler;
import org.tksj.bim.common.mq.MqConsumer;

import java.util.concurrent.CountDownLatch;

public class VertxMqConsumer<T> implements MqConsumer<T>
{
    private final MConsumerVertical<T> consumerVertical;

    private final CountDownLatch countDownLatch = new CountDownLatch(1);


    public VertxMqConsumer(Class<T> cls) {
        consumerVertical = new MConsumerVertical<>(cls);
        VertxManager.getVertx().deployVerticle(consumerVertical, stringAsyncResult -> countDownLatch.countDown());
        //等待初始化穿完成
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void consume(String topic, ConsumerHandler<T> handler) {
        consumerVertical.consume(topic, handler);
    }

    public static class MConsumerVertical<T> extends AbstractVerticle {

        private final Class<T> cls;
        private EventBus eventBus;

        public MConsumerVertical(Class<T> cls) {
            this.cls = cls;
        }

        @Override
        public void start() {
            eventBus = vertx.eventBus();
            eventBus.registerCodec(new BeanCodec<>(cls));
        }

        public void consume(String topic, ConsumerHandler<T> handler) {
            eventBus.consumer(topic, (Handler<Message<T>>) msg -> handler.handler(msg.body()));
        }
    }
}